package com.tangji.signaltower.runners.client;

import com.jsoniter.JsonIterator;
import com.jsoniter.any.Any;
import com.tangji.antenna.bean.EventFactory;
import com.tangji.antenna.bean.ProviderTypeEnum;
import com.tangji.signaltower.runners.WSClient;
import io.netty.channel.*;
import io.netty.handler.codec.http.FullHttpResponse;
import io.netty.handler.codec.http.websocketx.*;
import io.netty.util.CharsetUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * Created with IntelliJ IDEA.
 * User: Amarendar
 * Date: 2020/6/15
 * Time: 下午3:19
 */
public class ForexPoWebSocketClientHandler extends SimpleChannelInboundHandler<Object> {

    private static final Logger log = LoggerFactory.getLogger(ForexPoWebSocketClientHandler.class);

    WebSocketClientHandshaker handshaker;
    ChannelPromise handshakeFuture;

    public void handlerAdded(ChannelHandlerContext ctx) {
        this.handshakeFuture = ctx.newPromise();
    }

    public WebSocketClientHandshaker getHandshaker() {
        return handshaker;
    }

    public void setHandshaker(WebSocketClientHandshaker handshaker) {
        this.handshaker = handshaker;
    }

    public ChannelPromise getHandshakeFuture() {
        return handshakeFuture;
    }

    public void setHandshakeFuture(ChannelPromise handshakeFuture) {
        this.handshakeFuture = handshakeFuture;
    }

    public ChannelFuture handshakeFuture() {
        return this.handshakeFuture;
    }

    /**
     * 客户端连接服务器后被调用
     *
     * @param ctx
     */
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        log.info("连接建立成功");
    }

    /**
     * 服务端端终止连接 后触发此函数
     *
     * @param ctx
     */
    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        log.info("服务端终止了服务");
    }

    /**
     * 发生异常时被调用
     *
     * @param ctx
     * @param cause
     */
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        //捕捉异常信息
        log.error("服务端发生异常【" + cause.getMessage() + "】", cause);
        ctx.close();
    }

    /**
     * 接受到数据信息
     *
     * @param ctx
     * @param msg
     */
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, Object msg) throws Exception {
        log.info("接收到消息," + this.handshaker.isHandshakeComplete());
        Channel ch = ctx.channel();
        FullHttpResponse response;
        if (!this.handshaker.isHandshakeComplete()) {
            try {
                response = (FullHttpResponse) msg;
                //握手协议返回，设置结束握手
                this.handshaker.finishHandshake(ch, response);
                //设置成功
                this.handshakeFuture.setSuccess();
                log.info("WebSocket Client connected! response headers[sec-websocket-extensions]:{}", response.headers());
            } catch (WebSocketHandshakeException var7) {
                log.warn("error!", var7);
                FullHttpResponse res = (FullHttpResponse) msg;
                String errorMsg = String.format("WebSocket Client failed to connect,status:%s,reason:%s", res.status(), res.content().toString(CharsetUtil.UTF_8));
                this.handshakeFuture.setFailure(new Exception(errorMsg));
            }
        } else if (msg instanceof FullHttpResponse) {
            response = (FullHttpResponse) msg;
            //this.listener.onFail(response.status().code(), response.content().toString(CharsetUtil.UTF_8));
            throw new IllegalStateException("Unexpected FullHttpResponse (getStatus=" + response.status() + ", content=" + response.content().toString(CharsetUtil.UTF_8) + ')');
        } else {
            WebSocketFrame frame = (WebSocketFrame) msg;
            if (frame instanceof TextWebSocketFrame) {
                WSClient.msgNum++;
                if (WSClient.msgNum > 2000) {
                    WSClient.msgNum = 0;
                }
                //websocket的文本类消息都在这里处理
                TextWebSocketFrame textFrame = (TextWebSocketFrame) frame;
                String text = textFrame.text();
                log.info("==== 收到消息 TextWebSocketFrame,{}", text);
                try {
                    EventFactory.saveDataToQueue(ProviderTypeEnum.POLYGON, text, null);
//                    String type = body.get("type").toString();
//                    if ("ping".equalsIgnoreCase(type)) {
//                        ctx.channel().writeAndFlush(new TextWebSocketFrame("{\"type\":\"pong\"}"));
//                        log.info("自动回复:pang");
//                    } else {
//                        if (!EventFactory.saveDataToQueue(ProviderTypeEnum.POLYGON_FOREX, text, body)) {
//                            log.info("延迟超过5秒,尝试重连");
//                            ctx.channel().close();
//                        }
//                    }
                } catch (Exception e) {
                    log.warn("error JsonIterator check  !", e);
                }
            } else if (frame instanceof BinaryWebSocketFrame) {
                log.info("BinaryWebSocketFrame");
            } else if (frame instanceof PingWebSocketFrame) {
                log.info("WebSocket Client received ping");
                ctx.channel().writeAndFlush(new PongWebSocketFrame());
            } else if (frame instanceof PongWebSocketFrame) {
                log.info("WebSocket Client received pong");
            } else if (frame instanceof CloseWebSocketFrame) {
                log.info("receive close frame");
                ch.close();
            }

        }
    }

}